package com.cmbyc.JDBC.source

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.SourceFunction

import java.sql.DriverManager

/**
 *
 * @program: com.cmbyc.JDBC
 * @author: YCLW058
 * @create: 2021-05-19 13:57
 * @decsription:
 *
 * */

class Source3 extends MySQLSource {

  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/cd_mysql", "root", "root")
    val sql = "select s1.id,s1.name,s2.age from Student as s1 join Student3 as s2 on s1.id=s2.id"
    ps = conn.prepareStatement(sql)

  }

  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {
  val rs=ps.executeQuery()
    while (rs.next()){
      val id = rs.getInt("id")
      val name = rs.getString("name")
      val age = rs.getInt("age")
      ctx.collect(Student(id = id, name = name, age = age))
    }

  }

}
